package com.lagou.rpc.consumer.client;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Configuration
public class Registry {
    @Value("${zookeeper.host}")
    private String host;

    @Value("${zookeeper.port}")
    private String port;

    private CuratorFramework zkClient;

    private PathChildrenCache childrenCache;

    public void init() throws UnknownHostException {
        zkClient = getZkClient();
    }

    public void startWatch(ServerChangeProcessor processor) throws Exception {
        //注册监听
        childrenCache = new PathChildrenCache(zkClient, "/servers", false);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener((framework, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED:case CHILD_UPDATED:case CHILD_REMOVED:
                    List<ServerInfo> serverInfos = getServerInfos();
                    processor.onServerChange(serverInfos);
                    break;
                default:
                    break;
            }
        });
    }

    public List<ServerInfo> getServerInfos() {
        List<ServerInfo> serverInfos = new ArrayList<>();

        try {
            List<String> serverList = zkClient.getChildren()
                    .forPath("/servers");
            for (String serverItem : serverList) {
                String[] items = serverItem.split(":");

                ServerInfo serverInfo = new ServerInfo();
                serverInfo.setIp(items[0]);
                serverInfo.setPort(Integer.parseInt(items[1]));

                byte[] valueBytes = zkClient.getData().forPath(
                  "/servers/" + items[0] + ":" + items[1]
                );

                if (valueBytes == null) {
                    serverInfo.setLastTimeElapsed(0L);
                }
                else {
                    String strValue = new String(valueBytes);
                    serverInfo.setLastTimeElapsed(Long.parseLong(strValue));
                }

                serverInfos.add(serverInfo);
            }
        } catch (Exception ignored) {
        }
        return serverInfos;
    }

    private CuratorFramework getZkClient() throws UnknownHostException {
        RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(host + ":" + port)
                //.sessionTimeoutMs(10000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retry)
                .namespace("service")
                .build();
        client.start();
        return client;
    }

    public void clearServerStatics(ServerInfo serverInfo) {
        String path = "/servers/" + serverInfo.getIp() + ":" + serverInfo.getPort();
        try {
            zkClient.setData().forPath(path, "0".getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void updateServerStatics(ServerInfo serverInfo, ConsumeStatics statics) {
        String path = "/servers/" + serverInfo.getIp() + ":" + serverInfo.getPort();
        try {
            zkClient.setData().forPath(path, String.valueOf(statics.getLastTimeElapse()).getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
